package com.shanshan.wheel.rpc.common.remoting.transport.clent;

import com.shanshan.wheel.rpc.common.compress.CompressType;
import com.shanshan.wheel.rpc.common.registry.ServiceDiscovery;
import com.shanshan.wheel.rpc.common.remoting.constant.RpcMessageType;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcMessage;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcRequest;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcResponse;
import com.shanshan.wheel.rpc.common.remoting.transport.codec.RpcMessageDecoder;
import com.shanshan.wheel.rpc.common.remoting.transport.codec.RpcMessageEncoder;
import com.shanshan.wheel.rpc.common.serialize.SerializationType;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @Description
 * @ClassName NettyClient
 * @Author 山山大魔王
 * @Date 2022/9/5 11:19
 * @Version 1.0
 */
@Component
@Slf4j
public class NettyClient {

    private final NioEventLoopGroup group;

    Bootstrap bootstrap;

    @Autowired
    private ChannelProvider channelProvider;

    @Autowired
    private ServiceDiscovery serviceDiscovery;

    @Autowired
    private UnprocessedRequests unprocessedRequests;

    public NettyClient(){
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                // 设置连接超时时间5s
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcMessageDecoder());
                        pipeline.addLast(new RpcMessageEncoder());
                        // 给管道添加IdleStateHandler ，设置为5秒内没有写操作触发
                        pipeline.addLast(new IdleStateHandler(0,5,0, TimeUnit.SECONDS));
                        pipeline.addLast(new NettyRpcClientHandler());
                    }
                });
    }

    /**
     * 用于与 inetSocketAddress建立连接，并获取连接的Channel
     * @param inetSocketAddress
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    @SneakyThrows
    private Channel doConnect(InetSocketAddress inetSocketAddress) {
        // 创建CompletableFuture用于异步存取连接成功后的管道
        CompletableFuture<Channel> channelCompletableFuture = new CompletableFuture<>();
        // 连接 inetSocketAddress
        bootstrap.connect(inetSocketAddress)
                // 设置监听器，当操作完成后调用该监听器
                .addListener((ChannelFutureListener) channelFuture -> {
                    // 判断连接是否成功
                    if(channelFuture.isSuccess()){
                        log.info("客户端已连接 [{}] 成功！", inetSocketAddress.toString());
                        // 连接成功，将Channel存入completableFuture中
                        channelCompletableFuture.complete(channelFuture.channel());
                    }else {
                        throw new IllegalStateException("连接失败...");
                    }
                });
        // 返回连接成功得到的Channel，连接操作未完成会一直阻塞在此处
        return channelCompletableFuture.get();
    }

    /**
     * 获取inetSocketAddress的连接 （Channel对象），若没有则建立连接
     * @param inetSocketAddress
     * @return
     */
    public Channel getChannel(InetSocketAddress inetSocketAddress){
        // 先从channelProvider中获取channel
        Channel channel = channelProvider.get(inetSocketAddress);
        if(channel == null){
            // 如果没有获取到则建立连接
            channel = doConnect(inetSocketAddress);
            // 在channelProvider中添加inetSocketAddress的连接channel
            channelProvider.set(inetSocketAddress, channel);
        }
        // 返回channel
        return channel;
    }
    public void close(){
        group.shutdownGracefully();
    }

    /**
     * 消费者发送包含调用信息的rpcRequest（发送过程是异步的）
     * 返回一个CompletableFuture，后续通过该Future获取Rpc调用结果
     * 由于netty消息发送是异步的，返回Future是非阻塞的，因此该方法是非阻塞的
     * @param rpcRequest
     * @return
     */
    public CompletableFuture<RpcResponse<Object>> sendRpcRequest(RpcRequest rpcRequest){
        // 用于存放返回的结果1
        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        // 根据rpcRequest获取调用服务名
        String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName();

        // 根据服务名获取提供方的地址
        InetSocketAddress serviceAddress = serviceDiscovery.lookupService(rpcServiceName);
        // 与该地址建立连接
        Channel channel = getChannel(serviceAddress);
        if(channel.isActive()){
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            // 构建好RpcMessage
            RpcMessage rpcMsg = RpcMessage.builder()
                    .codec(SerializationType.KRYO.getCode())
                    .compress(CompressType.GZIP.getCode())
                    .messageType(RpcMessageType.REQUEST_TYPE.getCode())
                    .data(rpcRequest).build();

            // 发送rpcMessage并设置监听
            channel.writeAndFlush(rpcMsg).addListener((ChannelFutureListener)  channelFuture ->{
                if(channelFuture.isSuccess()){
                    log.info("客户端发送消息：[{}]",rpcMsg);
                }else {
                    // 若消息发送失败，则关闭通道，并在resultFuture中存入异常发生原因，后续在获取结果时可知发送失败
                    channelFuture.channel().close();
                    resultFuture.completeExceptionally(channelFuture.cause());
                    log.error("发送失败", channelFuture.cause());
                }
            });

        }else {
            throw new IllegalStateException("连接处于非活动状态！");
        }
        return resultFuture;

    }


}
